package com.gis507.mapreduce.AISToMySql;

import com.gis507.mapreduce.AISBean.AisClassifyBean;
import com.gis507.mapreduce.Utils.UtilsClass;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

@Slf4j
public class AISInterpolationReducer extends Reducer<Text, AisClassifyBean,AisClassifyBean, NullWritable> {

    //设定的最大时间
    private static final double MAXSTANDTIME = 200;
    private static final double MINSTANDTIME = 60;
    //设定的最大距离
    private static final double STANDSTANCE = 4000;
    private static int sum = 0;

    @Override
    protected void reduce(Text key, Iterable<AisClassifyBean> values, Context context) throws IOException, InterruptedException {



            //实例化一个列表，用于存放数据
            List<AisClassifyBean> aisList = new ArrayList<>(1000);

            //遍历values
            for (AisClassifyBean value : values) {

                //将value添加到列表
                aisList.add(UtilsClass.getNewAisClassifyBean(value));

            }

            log.info("aisList的长度是：" + aisList.size());

            //所有插值得到的数据放到一个列表中
            ArrayList<AisClassifyBean> allAisInterList = new ArrayList<>();

            //所有满足的数据
            ArrayList<AisClassifyBean> rightAisList = new ArrayList<>();

            //aisList是一个放有数据的列表
            for (int i = 0; i < aisList.size() - 2; i++) {

                //判断两点之间的差是否大于规定的最大时间差
                //java中取绝对值 Math.abs()
                if (Math.abs(aisList.get(i + 1).getUnixTime() - aisList.get(i).getUnixTime()) > MAXSTANDTIME ||
                        Math.abs(aisList.get(i + 1).getUnixTime() - aisList.get(i).getUnixTime()) < MINSTANDTIME) {

                    //判断是否距离超限，如果超限就插值，不超限就说明可能是停靠，就可以分段
                    if (UtilsClass.getStance(aisList.get(i), aisList.get(i + 1)) > STANDSTANCE &&
                            UtilsClass.getStance(aisList.get(i), aisList.get(i + 1)) < 10000) {

                        //往数据里插值
                        ArrayList<AisClassifyBean> aisInterList = UtilsClass.insertPoint(aisList.get(i), aisList.get(i + 1),STANDSTANCE);
                        //将本次插值的数据放到总的插值列表中
                        allAisInterList.addAll(aisInterList);

                        //sum
//                    sum += 1;
                    } else {
                        log.info("这个点应该是个停靠点：" + aisList.get(i));
                        i = i + 1;
                    }

                } else {
                    //满足的数据列表
                    rightAisList.add(UtilsClass.getNewAisClassifyBean(aisList.get(i)));
                    log.info("符合条件的添加成功一条数据");
                }
            }

            //汇总数据
            rightAisList.addAll(allAisInterList);

            //写出
            for (AisClassifyBean aisInterBean : rightAisList) {

                context.write(aisInterBean,NullWritable.get());
            }
    }
}
